package com.ficus.road.maintain.service;

import com.ficus.road.maintain.config.DiseaseConfig;
import com.ficus.road.maintain.core.model.Hit;
import com.ficus.road.maintain.dao.CarVideoRecordDao;
import com.ficus.road.maintain.dao.InspectPictureDao;
import com.ficus.road.maintain.dto.FrameResultDTO;
import com.ficus.road.maintain.process.DataProcessor;
import com.ficus.road.maintain.process.DataWriterProcessor;
import com.ficus.road.maintain.process.ReplayDataProcessor;
import com.ficus.road.maintain.util.AliyunOSSUtil;
import com.ficus.road.maintain.websocket.WebSocketManager;
import com.ficus.road.maintain.websocket.WebSocketReplayManager;
import ficus.car_demo.FrameMeta;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections4.CollectionUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Service
@Slf4j
public class DataProcessorManager {

    @Getter
    private final Map<String, DataProcessor> dataProcessorMap = new ConcurrentHashMap<>();

    private final Map<String, ReplayDataProcessor> LiveProcessorMap = new ConcurrentHashMap<>();

    @Getter
    private final Map<String, ReplayDataProcessor> replayDataProcessorMap = new ConcurrentHashMap<>();

    private List<DataWriterProcessor> dataWriterProcessors;
    @Value("${processor-config.hit-processor-count}")
    private int threadCount;

    @Autowired
    private AliyunOSSUtil aliyunOSSUtil;
    @Autowired
    private WebSocketManager webSocketManager;

    @Autowired
    private WebSocketReplayManager webSocketReplayManager;
    @Autowired
    private HitService hitService;
    @Autowired
    private CarService carService;
    @Autowired
    private GpsHistoryService gpsHistoryService;
    @Value("${hit-config.max-distance}")
    private int maxDistance;
    @Value("${hit-config.gps-save-duration}")
    private int gpsSaveDuration;
    @Value("${hit-config.gps-minute-save-duration}")
    private int gpsMinuteSaveDuration;
    @Value("${hit-config.trip-save-duration}")
    private int tripSaveDuration;
    @Value("${hit-config.check-enable}")
    private boolean hitCheck;

    @Value("${delay-time-seconds}")
    private int delayTime;
    @Value("${rateLimiter.threshold}")
    private double rateLimiterThreshold;

    @Autowired
    private DiseaseConfig diseaseConfig;

    @Autowired
    private CarVideoRecordDao carVideoRecordDao;

    @Autowired
    private InspectPictureDao inspectPictureDao;

    @Autowired
    private CompanyService companyService;

    public DataProcessorManager() {

    }

    @PostConstruct
    void init() {
        dataWriterProcessors = new ArrayList<>(threadCount);
        for (int i = 0; i < threadCount; i++) {
            DataWriterProcessor dataWriterProcessor = new DataWriterProcessor("dataWriterProcessor" + i, aliyunOSSUtil, carVideoRecordDao, inspectPictureDao, this);
            dataWriterProcessor.start();
            dataWriterProcessors.add(dataWriterProcessor);
        }
    }

    /**
     * VideoReplayThread线程把需要播放的帧一次性压入ReplayDataProcessor的blockingQueue上
     *
     * @param request
     * @param picturePath
     * @param nextPicturePath
     * @param compressed
     * @param nextCompressed
     * @param uuid
     */
    public void putReplayFrame(FrameMeta.FrameResult request, String picturePath, String nextPicturePath, Boolean compressed, Boolean nextCompressed, String uuid) {
        try {
            ReplayDataProcessor replayDataProcessor = replayDataProcessorMap.get(uuid);
            if (replayDataProcessor == null) {
                if(request.getCarId().equals("JINXIUDONGLU")){
                    rateLimiterThreshold = 10;
                }
                replayDataProcessor = new ReplayDataProcessor(webSocketReplayManager, webSocketManager, diseaseConfig, maxDistance, false, hitService, false, delayTime, rateLimiterThreshold);
                replayDataProcessorMap.put(uuid, replayDataProcessor);
                new Thread(replayDataProcessor, "replayDataProcessor-" + uuid).start();
            }
            replayDataProcessor.put(new FrameResultDTO(uuid, picturePath, nextPicturePath, compressed, nextCompressed, request));
        } catch (InterruptedException e) {
            log.error("InterruptedException", e);
        }
    }

    public void closeReplayWs(String uuid) {
        ReplayDataProcessor replayDataProcessor = replayDataProcessorMap.get(uuid);
        if (replayDataProcessor != null) {
            replayDataProcessor.closeWs(uuid);
        }
    }

    public void putLiveReplay(FrameMeta.FrameResult request, String picturePath, String nextPicturePath, Boolean compressed, Boolean nextCompressed, String uuid) {
        try {
            ReplayDataProcessor replayDataProcessor = LiveProcessorMap.get(uuid);
            if (replayDataProcessor == null) {
                replayDataProcessor = new ReplayDataProcessor(webSocketReplayManager, webSocketManager, diseaseConfig, maxDistance, false, hitService, true, delayTime, rateLimiterThreshold);
                LiveProcessorMap.put(uuid, replayDataProcessor);
                new Thread(replayDataProcessor, "liveDataProcessor-" + uuid).start();
            }
            replayDataProcessor.put(new FrameResultDTO(uuid, picturePath, nextPicturePath, compressed, nextCompressed, request));
        } catch (InterruptedException e) {
            log.error("InterruptedException", e);
        }
    }

    public void putReplayHits(String uuid, Map<Long, List<Hit>> hitsMap, Long time) {
        ReplayDataProcessor replayDataProcessor = replayDataProcessorMap.get(uuid);
        if (replayDataProcessor == null) {
            replayDataProcessor = new ReplayDataProcessor(webSocketReplayManager, webSocketManager, diseaseConfig, maxDistance, false, hitService, false, delayTime, rateLimiterThreshold);
            replayDataProcessorMap.put(uuid, replayDataProcessor);
            new Thread(replayDataProcessor, "replayDataProcessor-" + uuid).start();
        }
        replayDataProcessor.putReplayHits(hitsMap, time);
    }

    //实时数据处理、数据落地处理
    public void put(FrameMeta.SendMetaRequest sendRequest) {
        List<FrameMeta.FrameResult> list = sendRequest.getFramesList();
        if (CollectionUtils.isNotEmpty(list)) {
            String carId = list.get(0).getCarId();
            //拿到carId对应的线程
            DataProcessor dataProcessor = dataProcessorMap.get(carId);
            if (dataProcessor == null) {
                dataProcessor = new DataProcessor(hitService, carService,
                        gpsHistoryService, companyService, diseaseConfig, maxDistance, gpsSaveDuration, gpsMinuteSaveDuration, tripSaveDuration, hitCheck);
                // dataProcessorMap最多存放线程数量和car的数量相同, 且线程一旦创建, 没有数据过来的时候, 线程会在blockingDeque的take方法上一直处于waiting状态
                dataProcessorMap.put(carId, dataProcessor);
                //实时数据处理
                new Thread(dataProcessor, "dataProcessor-" + carId).start();
            }

            //为什么要保证同一辆车上传的数据由同一个线程来处理？用不同线程处理一辆车上传上来的数据，会有线程安全问题，无法保证数据顺序存储？或者是为了分摊各个线程的压力？
            int index = carId.hashCode() & Integer.MAX_VALUE % threadCount;
            try {
                //根据carId获取oss处理器编号,将FrameResult放入dataWriterProcessor中的阻塞队列
                //dataWriterProcessor会将数据上传到OSS
                dataWriterProcessors.get(index).put(sendRequest);
            } catch (InterruptedException e) {
                log.error("InterruptedException", e);
            }

        }

    }
}
